import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Random;
import java.util.stream.Stream;

/**
 * 20G的文件，每行存放1个URL，请用1G的内存计算出重复次数最多的前100个URL
 * 
 * @author Atomic
 *
 */
public class PaTest {

    public static final char[] CHARSANDNUM = {//'0','1','2','3','4','5','6','7','8','9',
                                        'A','B','C','D','E','F','G','H','I','J','K','L','M','N','O','P','Q','R','S','T','U','V','W','X','Y','Z',
                                        'a','b','c','d','e','f','g','h','i','j','k','l','m','n','o','p','q','r','s','t','u','v','w','x','y','z'};
    
    /**
     * 测试文件大小
     */
    public static final long FILESIZE = 4L*1024*1024*1024;
    
    public static final long READSIZE = 10*1024*1024;
    
    public static final int AVG_STRLEN = 6;
    
    public static Random rd = new Random();
    
    public static void main(String[] args) {
        //测试数据所在文件
        String filePath="D:\\Download\\patest\\patest.txt";
        //生成测试数据
        creatTestDate(filePath);
        long beginTime = System.currentTimeMillis();
        try {
            //统计
            Map<String,Integer> resMap = getTop100V3(filePath);
            //打印结果
            print(resMap);
        } catch (Exception e) {
            e.printStackTrace();
        }
        long useTime = (System.currentTimeMillis()-beginTime)/(1000);
        System.out.println("总计用时(s)："+useTime);
        
    }
    
    /**
     * 生成测试数据: 全是数字，如果数字是1，则重复1次，如果是2，则重复出现2次，如果是n，则重复出现m*(n%100)次,m为正整数
     * 
     * @param filePath
     * @param fileSize 保存测试数据的文件大小。当小于等于0时，默认为2*1024*1024*1024个字符
     */
    public static void creatTestDate(String filePath) {
        System.out.println("生成测试数据");
        if (Files.exists(Paths.get(filePath))) {
            return;
        }
        long fs = FILESIZE;
        try (BufferedWriter out = Files.newBufferedWriter(Paths.get(filePath));) {
            int count = rd.nextInt(100000000);
            while (fs > 0) {
                String tmpStr = count + "\n";
                int cfCount = count%100;
                for (int i = 0; i < cfCount && fs > 0; i++) {
                    fs -= tmpStr.length();
                    out.write(tmpStr);
                }
                count = rd.nextInt(100000000);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    
    private static void print(Map<String,Integer> map) {
        List<Entry<String,Integer>> entryArr = new ArrayList<Entry<String,Integer>>(map.size());
        entryArr.addAll(map.entrySet());
        
        entryArr.sort(new Comparator<Entry<String,Integer>>() {
            @Override
            public int compare(Entry<String, Integer> o1, Entry<String, Integer> o2) {
                return o1.getValue()-o2.getValue();
            }
        });
        
        for(int i=0;i<entryArr.size();i++) {
            Entry<String,Integer> entry = entryArr.get(i);
            System.out.println(i+"："+entry.getKey()+":"+entry.getValue());
        }
    }
    
    /**
     * 直接统计
     * 
     * @param filePath
     * @return
     * @throws Exception
     */
    public static Map<String,Integer> getTop100V1(String filePath) throws Exception {
        final int sep=10000000;
        Map<String,Integer> subMap = new HashMap<String,Integer>((int) (READSIZE/AVG_STRLEN));
        List<String> tmpFileList = new ArrayList<String>();
        tmpFileList.add(filePath);
        Map<String,Integer> resMap = new HashMap<String,Integer>(200);
        
        for(int i=0;i<tmpFileList.size();i++) {
            long beginTime2 = System.currentTimeMillis();
            System.out.println("正在处理文件："+tmpFileList.get(i));
            BufferedReader in = Files.newBufferedReader(Paths.get(tmpFileList.get(i)));
            BufferedWriter out = null;
            String line = in.readLine();
            long currentSize=0;
            boolean flag=false;
            long lineCount=0;
            while(line!=null) {
                if(subMap.containsKey(line)) {
                    subMap.put(line, subMap.get(line)+1);
                }else if(!flag&&currentSize<READSIZE) {
                    subMap.put(line, 1);
                }else {
                    flag=true;
                    if(out==null) {
                        String tmpFilePath = filePath.replace(".txt", "_"+tmpFileList.size()+".txt");
                        tmpFileList.add(tmpFilePath);
                        out = Files.newBufferedWriter(Paths.get(tmpFilePath));
                    }
                    out.write(line+"\n");
                }
                lineCount++;
                currentSize+=line.length();
                if(lineCount%sep==0) {
                    System.out.println("处理到第"+lineCount/sep+"个"+sep+"行");
                }
                line = in.readLine();
            }
            System.out.println("处理了"+lineCount+"行");
            if(out!=null) {
                out.close();
            }
            if(in!=null) {
                in.close();
            }
            resMap.putAll(getTop100FromMap(subMap));
            subMap.clear();
            if(resMap.size()>100) {
                resMap = getTop100FromMap(resMap);
            }
            long useTime2 = (System.currentTimeMillis()-beginTime2)/1000;
            System.out.println("处理文件："+tmpFileList.get(i)+"，耗时(s)："+useTime2+",字符数："+currentSize);
        }
        for(int i=1;i<tmpFileList.size();i++) {
            Files.deleteIfExists(Paths.get(tmpFileList.get(i)));
        }
        resMap = getTop100FromMap(resMap);
        return resMap;
    }
    
    private static  Map<String,Integer> getTop100FromMap(Map<String,Integer> map){
        Map<String,Integer> resMap = new HashMap<String,Integer>(200);
        if(map.size()<100) {
            resMap.putAll(map);
            return resMap;
        }
        List<Entry<String,Integer>> heapList = new ArrayList<Entry<String,Integer>>();
        for(Entry<String,Integer> ety:map.entrySet()) {
            heapList.add(ety);
            shiftUp(heapList,heapList.size()-1);
        }
        for(int i=0;i<100;i++) {
            resMap.put(heapList.get(0).getKey(),heapList.get(0).getValue());
            heapList.set(0, heapList.get(heapList.size()-1));
            heapList.remove(heapList.size()-1);
            shiftDown(heapList, 0);
        }
        
        return resMap;
    }
    
    private static void shiftUp(List<Entry<String,Integer>> heapList,int currentIdx) {
        int topIdx = currentIdx%2==1?(currentIdx-1)/2:(currentIdx-2)/2;
        if(topIdx<0) {
            return;
        }
        if(heapList.get(topIdx).getValue()>=heapList.get(currentIdx).getValue()) {
            return;
        }
        Entry<String,Integer> tmpVal = heapList.get(topIdx);
        heapList.set(topIdx, heapList.get(currentIdx));
        heapList.set(currentIdx, tmpVal);
        shiftUp(heapList, topIdx);
    }
    
    private static void shiftDown(List<Entry<String,Integer>> heapList,int currentIdx) {
        int leftDownIdx = 2*currentIdx+1;
        int rightDownIdx = 2*currentIdx+2;
        
        if(leftDownIdx>=heapList.size()) {
            return;
        }
        Entry<String,Integer> currentEty = heapList.get(currentIdx);
        Entry<String,Integer> leftEty = heapList.get(leftDownIdx);
        
        if(rightDownIdx>=heapList.size()) {
            if(currentEty.getValue()>=leftEty.getValue()) {
                return;
            }else {
                Entry<String,Integer> tmp = heapList.get(leftDownIdx);
                heapList.set(leftDownIdx, heapList.get(currentIdx));
                heapList.set(currentIdx, tmp);
                shiftDown(heapList, leftDownIdx);
                return;
            }
        }
        Entry<String,Integer> rightEty = heapList.get(rightDownIdx);
        if(currentEty.getValue()>leftEty.getValue()&&currentEty.getValue()>rightEty.getValue()) {
            return;
        }else if(leftEty.getValue()>currentEty.getValue()&&leftEty.getValue()>rightEty.getValue()) {
            Entry<String,Integer> tmp = heapList.get(leftDownIdx);
            heapList.set(leftDownIdx, heapList.get(currentIdx));
            heapList.set(currentIdx, tmp);
            shiftDown(heapList, leftDownIdx);
            return;
        }else {
            Entry<String,Integer> tmp = heapList.get(rightDownIdx);
            heapList.set(rightDownIdx, heapList.get(currentIdx));
            heapList.set(currentIdx, tmp);
            shiftDown(heapList, rightDownIdx);
            return;
        }
    }
    
    /**
     * 将大文件分成一个一个小文件之后，先在小文件中进行统计，然后汇总统计
     * 
     * @param filePath
     * @return
     * @throws Exception
     */
    public static Map<String,Integer> getTop100V2(String filePath) throws Exception {
        String[] subFilePathArr = splitFile(filePath, (int) (FILESIZE/READSIZE));
        Map<String,Integer> resMap = new HashMap<String,Integer>(200);
        for(String subFilePath:subFilePathArr) {
            resMap.putAll(getTop100V1(subFilePath));
            if(resMap.size()>100) {
                resMap=getTop100FromMap(resMap);
            }
        }
        for(String f:subFilePathArr) {
            Files.deleteIfExists(Paths.get(f));
        }
        return resMap;
    }
    
    private static String[] splitFile(final String filePath,final int fileNum) throws Exception{
        long beginTime = System.currentTimeMillis();
        System.out.println("分割文件，子文件数为："+fileNum);
        String[] subFilePathArr = new String[fileNum]; 
        BufferedWriter[] out  = new BufferedWriter[fileNum];
        for(int i=0;i<fileNum;i++) {
            subFilePathArr[i] = filePath.replace(".txt", "_sub"+i+".txt");
            out[i]=Files.newBufferedWriter(Paths.get(subFilePathArr[i]));
        }
        BufferedReader in = Files.newBufferedReader(Paths.get(filePath));
        String line = in.readLine();
        while(line!=null) {
            String wLine = line+"\n";
            out[Math.abs(wLine.hashCode())%fileNum].write(wLine);
            line = in.readLine();
        }
        in.close();
        for(BufferedWriter bw:out) {
            bw.close();
        }
        long useTime = (System.currentTimeMillis()-beginTime)/1000;
        System.out.println("分割文件用时(s)："+useTime);
        return subFilePathArr;
    }
    
    /**
     * 在getTop100V2的基础上，引入多线程处理
     * 
     * @param filePath
     * @return
     * @throws Exception
     */
    public static Map<String,Integer> getTop100V3(String filePath) throws Exception {
        String[] subFilePathArr = splitFile(filePath, (int) (FILESIZE/READSIZE));
        Map<String,Integer> resMap = new HashMap<String,Integer>(200);
        Stream<String> stream = Stream.of(subFilePathArr);
        stream.parallel().forEach(x->{
            try {
                Map<String,Integer> subMap = getTop100V1(x);
                synchronized(resMap){
                    resMap.putAll(subMap);
                    if(resMap.size()>100) {
                        Map<String,Integer> tmpMap = getTop100FromMap(resMap);
                        resMap.clear();
                        resMap.putAll(tmpMap);
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
        for(String f:subFilePathArr) {
            Files.deleteIfExists(Paths.get(f));
        }
        return resMap;
    }
}
